package com.example.rocketmq.test1;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.CountDownLatch;

/*
生产者发送消息的方式：
   1、单向发送 sendOneway
   2、同步发送 send
       向队列中发送消息，并且得到队列的响应才会继续往下执行，否则会阻塞
   3、异步发送 send(msg, new SendCallBack() {...} )
       通过异步发送内置的回调函数，自行指定回调处理
*/
public class Producer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        //初始化⼀个消息⽣产者
        DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
        // 指定nameserver地址
        producer.setNamesrvAddr("10.100.154.15:9800;10.100.154.16:9800");
        // 启动消息⽣产者服务
        producer.start();
        int threadCount = 2;
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 2; i < 4; i++) {
            try {
                // 创建消息。消息由Topic,Tag和body三个属性组成，其中Body就是消息内容
                Message msg = new Message("TopicZTKTest2","TagA",("Hello RocketMQ "
                        +i).getBytes(RemotingHelper.DEFAULT_CHARSET));
//                //发送消息，获取发送结果
//                SendResult sendResult = producer.send(msg);

                // 单向发送
                // producer.sendOneway(msg);
                // System.out.println("消息发送成功！！！");
                // 同步发送
                // SendResult sendResult = producer.send(msg);
                // System.out.printf("这个符号是什么意思？%s%n", sendResult);
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.println(sendResult.getSendStatus());
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        countDownLatch.countDown();
                        throwable.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
        countDownLatch.await();
        //消息发送完后，停⽌消息⽣产者服务。
        producer.shutdown();
    }
}
